package workpool

import (
	"errors"
	"log"
	"strconv"
	"sync"
)

//Dispatcher 调度器 负责将Job分发给不同的Worker进行处理
type Dispatcher struct {
	workPool   chan JobChan //Worker实际的工作队列会注册到该缓冲池以供调度
	async      bool         //是否异步调度
	maxWorkers int          //最大工作者数量
	maxJobs    int          //最大任务数量
	jobs       JobChan      //待调度的任务队列
	workers    []*worker
	activated  bool
	lock       sync.RWMutex
}

//NewDispatcher 创建一个任务调度器
func NewDispatcher(options ...Option) (*Dispatcher, error) {
	setting := &dispatcherSetting{
		async:      false,
		maxWorkers: 10,
		maxJobs:    1000,
	}
	for _, v := range options {
		v(setting)
	}
	if setting.maxJobs <= 0 {
		return nil, errors.New("MaxJobs必须大于0")
	}
	if setting.maxWorkers <= 0 {
		return nil, errors.New("MaxWorkers必须大于0")
	}

	pool := make(chan JobChan, setting.maxWorkers)
	jobs := make(JobChan, setting.maxJobs)
	return &Dispatcher{
		workPool:   pool,
		async:      setting.async,
		maxWorkers: setting.maxWorkers,
		maxJobs:    setting.maxJobs,
		jobs:       jobs,
	}, nil
}

//Run 启动调度器 除非stop否者无法重复Run
func (d *Dispatcher) Run() error {
	d.lock.Lock()
	defer d.lock.Unlock()
	if d.activated {
		return errors.New("不允许重复启动调度器")
	}
	// 开始运行 n 个 worker
	for i := 1; i <= d.maxWorkers; i++ {
		w := newWorker(strconv.Itoa(i), d.workPool)
		d.workers = append(d.workers, w)
		w.start()
	}
	go d.dispatch()
	d.activated = true
	return nil
}

//HandleJob 将job交给调度器处理
//若同步调度,则当无空闲Worker时该方法阻塞
//建议Run之后再调用HandleJob，否者可能无法及时调度甚至死锁
func (d *Dispatcher) HandleJob(job Job) {
	d.jobs <- job
}

func (d *Dispatcher) Stop() {
	d.lock.Lock()
	defer d.lock.Unlock()
	if !d.activated {
		return
	}
	wg := sync.WaitGroup{}
	wg.Add(len(d.workers))
	for _, v := range d.workers {
		go func(w *worker) {
			w.stop()
			wg.Done()
		}(v)
	}
	wg.Wait()
	d.workers = make([]*worker, 0)
	d.activated = false
}

func (d *Dispatcher) dispatch() {
	dispatch := func(job Job) {}
	if !d.async {
		dispatch = d.syncDispatch
	} else {
		dispatch = d.asyncDispatch
	}
	for {
		select {
		case job := <-d.jobs:
			dispatch(job)
		}
	}
}

func (d *Dispatcher) asyncDispatch(job Job) {
	//这里采用开启新协程的方式，保证调度不会阻塞影响Job收集
	//虽然也是每个Job都会开启一个协程，但由于不直接处理业务逻辑，所以生命周期短暂 不会产生大量协程
	log.Println("异步调度Job：", job.GetId())
	go func(job Job) {
		// 尝试获取一个可用的 工作者Job队列，这里会阻塞直到有可用的 worker
		jobChannel := <-d.workPool
		// 获取到之后 将任务交给该工作者的Job队列
		jobChannel <- job
	}(job)
}

func (d *Dispatcher) syncDispatch(job Job) {
	//同步调度的话，一旦所有worker都处于业务处理中，则<-d.WorkerPool将阻塞,反之导致JobQueue满后JobQueue <- job阻塞
	log.Println("同步调度Job：", job.GetId())
	// 尝试获取一个可用的 工作者Job队列，这里会阻塞直到有可用的 worker
	jobChannel := <-d.workPool
	// 获取到之后 将任务交给该工作者的Job队列
	jobChannel <- job
}

type dispatcherSetting struct {
	async      bool //是否异步调度
	maxWorkers int  //最大工作者数量
	maxJobs    int  //最大任务数量
}

type Option func(setting *dispatcherSetting)

//MaxJobs 最大任务数
func MaxJobs(maxJobs int) Option {
	return func(setting *dispatcherSetting) {
		setting.maxJobs = maxJobs
	}
}

//MaxWorkers 最大工作者数
func MaxWorkers(maxWorkers int) Option {
	return func(setting *dispatcherSetting) {
		setting.maxWorkers = maxWorkers
	}
}

//Async 是否异步调度，不建议单个job执行时间长且数量过大的情况采用异步调度，否者会创建大量协程
func Async(async bool) Option {
	return func(setting *dispatcherSetting) {
		setting.async = async
	}
}
